package JJJJJJava.hivedatasource;

import javafx.beans.property.ReadOnlyBooleanWrapper;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.codehaus.janino.Java;

/**
 * @Auther: Mengkunxuan
 * @Date:2018/9/299:55
 * @Description:
 */
public class HiveDataSource {
    @SuppressWarnings("deprecation")
    public static void main(String[] args) {
        //首先还是创建SparkConf
        SparkConf conf  = new SparkConf().setAppName("HiveDataSource").setMaster("local");
        //创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        //创建HiveContext,注意他接受的是SparkContext作为参数,不是JavaSparkContext
        //sc()方法应当是将javasc转换为SparkContext类型返回
        HiveContext hiveContext = new HiveContext(sc.sc());
        //第一个功能,使用HiveContext的sql()方法,可以执行Hive中能狗执行的HiveSQL语句
        //判断是否存在student_info表,如果存在则删除
        hiveContext.sql("drop table if exists student_infos");
        //判断student_info表是否不存在,如果不存在,则创建该表
        hiveContext.sql("CREATE TABLE IF NOT EXISTS student_infos (name STRING, age int)");
        // 将学生基本信息数据导入student_infos表
        hiveContext.sql("load data local inpath 'C:\\Users\\Administrator\\Desktop\\第一阶段代码\\第81讲-Spark SQL：Hive数据源复杂综合案例实战\\文档\\student_infos.txt' into table student_infos");
        //用同样的方式给student_scores导入数据
        hiveContext.sql("drop table if exists student_scores");
        hiveContext.sql("create table if not exists student_scores (name STRING,score int)");
        hiveContext.sql("LOAD data local inpath 'C:\\Users\\Administrator\\Desktop\\第一阶段代码\\第81讲-Spark SQL：Hive数据源复杂综合案例实战\\文档\\student_scores.txt' into table student_scores");
        //第二个功能,执行sql还可以返回DataFrame,用于查询
        //执行sql查询,关联两张表,查询成绩大于80分的学生
        DataFrame goodStudentsDF = hiveContext.sql("select si.name ,si.age,ss.score " +
                "from student_infos si " +
                "join student_scores ss on si.name =ss.name " +
                "where ss.score>=80");
        //第三个功能 ,可以将DataFrame中的数据,理论上来说,DataFrame对应的RDD的元素,是Row即可
        //将DataFrame中的数据保存到hive表中
        //将DataFrame中的数据保存到good_student_infos表中
        hiveContext.sql("drop table if exists good_student_infos");
        goodStudentsDF.saveAsTable("good_student_infos");
        // 第四个功能,可以用table()方法,针对hive表,直接创建DataFrame
        //然后针对good_student_infos表,直接创建DataFrame
        Row[] goodStudentRows = hiveContext.table("good_student_infos").collect();
        for (Row goodStudentRow:goodStudentRows){
            System.out.println(goodStudentRow);
        }
        sc.close();

    }
}
